package com.atguigu.chapter06;

import com.atguigu.Bean.OrderEvent;
import com.atguigu.Bean.TxEvent;
import lombok.val;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.HashSet;

/**
 * @ClassName: Flink03_Order_Match
 * @Description:
 * @Author: kele
 * @Date: 2021/4/6 13:50
 *
 * 6.4	订单支付实时监控
 *
 * 需求：来自两条流的订单交易匹配
 **/

public class Flink03_Order_Match {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        //获取订单信息
        SingleOutputStreamOperator<OrderEvent> order = env.readTextFile("in/OrderLog.csv")
                .flatMap(new FlatMapFunction<String, OrderEvent>() {
                    @Override
                    public void flatMap(String value, Collector<OrderEvent> out) throws Exception {

                        String[] info = value.split(",");

                        if ("pay".equals(info[1])) {
                            OrderEvent orderEvent = new OrderEvent(
                                    Long.valueOf(info[0]),
                                    info[1],
                                    info[2],
                                    Long.valueOf(info[3]));

                            out.collect(orderEvent);
                        }

                    }
                });


        //获取收据
        SingleOutputStreamOperator<TxEvent> rece = env.readTextFile("in/ReceiptLog.csv")
                .map(data -> {

                    String[] info = data.split(",");

                    return new TxEvent(info[0],
                            info[1],
                            Long.valueOf(info[2]));
                });


        //将两个流进行合并
        order.connect(rece)
                .keyBy(new KeySelector<OrderEvent, String>() {

                           @Override
                           public String getKey(OrderEvent value) throws Exception {
                               return value.getTxId();
                           }
                       },
                        new KeySelector<TxEvent, String>() {
                            @Override
                            public String getKey(TxEvent value) throws Exception {
                                return value.getTxId();
                            }
                        })
                .process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {

                    private HashMap<String, OrderEvent> order = new HashMap<>();
                    private HashMap<String, TxEvent> rece = new HashMap<>();

                    //针对第一个流的处理过程
                    @Override
                    public void processElement1(OrderEvent value, Context ctx, Collector<String> out) throws Exception {

                        if (rece.containsKey(value.getTxId())) {
                            out.collect("订单: " + value.getOrderId() + " 对账成功!");
                        }
                        else
                            order.put(value.getTxId(),value);
                    }

                    //针对第二个流的处理过程
                    @Override
                    public void processElement2(TxEvent value, Context ctx, Collector<String> out) throws Exception {
                        if (order.containsKey(value.getTxId())) {
                            out.collect("订单: " + order.get(ctx.getCurrentKey()).getOrderId() + " 对账成功!");
                        }
                        else
                            rece.put(value.getTxId(),value);
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
